package com.openness.crawler.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.coprocessor.AggregateProtocol;
import org.apache.hadoop.hbase.util.Bytes;

import com.openness.crawler.crawler.CrawlConfig;
import com.openness.crawler.util.Constants;

/**
 * @author Rain
 * @version 2013-5-22 上午9:57:14
 */

public class HBaseUtil {

	public static HTableDescriptor getHTableDescriptor(String tableName) {
		if (tableName == null || tableName.isEmpty()) {
			return null;
		}

		HTableDescriptor tableDesc = new HTableDescriptor(tableName);

		tableDesc
				.addFamily(new HColumnDescriptor(Constants.HBASE_TABLE_FAMILY));

		return tableDesc;
	}

	public static long getMaximumLongValueInFirstQualifier(String tableName,
			byte[] startRow, byte[] stopRow) throws IOException, Throwable {
		if (tableName == null || tableName.isEmpty()) {
			return Long.MIN_VALUE;
		}

		HTableInterface table = null;

		try {
			table = HBaseTablePool.get(tableName);

			Map<byte[], Long> result = table.coprocessorExec(
					AggregateProtocol.class, startRow, stopRow,
					new Batch.Call<AggregateProtocol, Long>() {

						@Override
						public Long call(AggregateProtocol protocol)
								throws IOException {
							Scan scan = new Scan();

							scan.addFamily(Constants.HBASE_TABLE_FAMILY);

							return protocol.getMax(new LongColumnInterpreter(),
									scan);
						}

					});

			long max = Long.MIN_VALUE;

			for (Entry<byte[], Long> entry : result.entrySet()) {
				long value = entry.getValue();

				if (value > max) {
					max = value;
				}
			}

			return max;
		} finally {
			if (table == null) {
				HBaseTablePool.put(table);
			}
		}
	}

	public static long getRowCount(String tableName, byte[] startRow,
			byte[] stopRow) throws IOException, Throwable {
		if (tableName == null || tableName.isEmpty()) {
			return 0;
		}

		HTableInterface table = null;

		try {
			table = HBaseTablePool.get(tableName);

			Map<byte[], Long> result = table.coprocessorExec(
					AggregateProtocol.class, startRow, stopRow,
					new Batch.Call<AggregateProtocol, Long>() {

						@Override
						public Long call(AggregateProtocol protocol)
								throws IOException {
							Scan scan = new Scan();

							scan.addFamily(Constants.HBASE_TABLE_FAMILY);

							return protocol.getRowNum(
									new LongColumnInterpreter(), scan);
						}

					});

			long rowCount = 0;

			if (result != null && result.size() > 0) {
				for (Entry<byte[], Long> entry : result.entrySet()) {
					rowCount += entry.getValue();
				}
			}

			return rowCount;
		} finally {
			if (table != null) {
				HBaseTablePool.put(table);
			}
		}
	}

	private static Map<String, byte[]> resultToMap(Result result) {
		if (result == null) {
			return null;
		}

		int size = result.size();

		Map<String, byte[]> map = new HashMap<String, byte[]>(size);

		for (KeyValue kv : result.raw()) {
			map.put(Bytes.toString(kv.getQualifier()), kv.getValue());
		}

		return map;
	}

	public static Map<String, byte[]> getRow(String tableName, byte[] row)
			throws IOException {
		if (tableName == null || tableName.isEmpty() || row == null
				|| row.length <= 0) {
			return null;
		}

		HTableInterface table = null;

		try {
			table = HBaseTablePool.get(tableName);

			Result result = table.get(new Get(row));

			int size = result.size();

			if (size == 0) {
				return null;
			} else {
				return resultToMap(result);
			}
		} finally {
			if (table != null) {
				HBaseTablePool.put(table);
			}
		}
	}

	public static List<Map<String, byte[]>> getRows(String tableName,
			List<byte[]> rows) throws IOException {
		if (tableName == null || tableName.isEmpty() || rows == null
				|| rows.size() <= 0) {
			return null;
		}

		List<Map<String, byte[]>> results = new ArrayList<Map<String, byte[]>>();

		for (int index = 0; index < rows.size(); index++) {
			results.add(null);
		}

		HTableInterface table = null;

		try {
			table = HBaseTablePool.get(tableName);

			List<Get> gets = new ArrayList<Get>();

			for (byte[] row : rows) {
				gets.add(new Get(row));
			}

			Result[] resArray = table.get(gets);

			if (resArray != null && resArray.length == rows.size()) {
				for (int index = 0; index < resArray.length; index++) {
					results.set(index, resultToMap(resArray[index]));
				}
			}

			return results;
		} finally {
			if (table != null) {
				HBaseTablePool.put(table);
			}
		}
	}

	public static List<Row> getFirstRows(String tableName, int max)
			throws IOException {
		List<Row> rows = new ArrayList<Row>();

		if (tableName == null || tableName.isEmpty() || max <= 0) {
			return rows;
		}

		HTableInterface table = null;

		ResultScanner rs = null;

		try {
			table = HBaseTablePool.get(tableName);

			Scan scan = new Scan();

			rs = table.getScanner(scan);

			Result[] results = rs.next(max);

			if (results != null && results.length > 0) {
				for (Result result : results) {
					Row row = new Row(result.getRow());

					for (KeyValue kv : result.raw()) {
						row.add(Bytes.toString(kv.getQualifier()),
								kv.getValue());
					}

					rows.add(row);
				}
			}

			return rows;
		} finally {
			if (rs != null) {
				rs.close();
			}

			if (table != null) {
				HBaseTablePool.put(table);
			}
		}

	}

	public static void putRow(String tableName, byte[] row, String qualifier,
			byte[] value) throws IOException {
		if (tableName == null || tableName.isEmpty() || row == null
				|| row.length <= 0 || qualifier == null || qualifier.isEmpty()
				|| value == null || value.length <= 0) {
			return;
		}

		HTableInterface table = null;

		try {
			table = HBaseTablePool.get(tableName);

			Put put = new Put(row);

			put.add(Constants.HBASE_TABLE_FAMILY, Bytes.toBytes(qualifier),
					value);

			table.put(put);
		} finally {
			if (table != null) {
				HBaseTablePool.put(table);
			}
		}
	}

	public static void putRow(String tableName, byte[] row,
			Map<String, byte[]> qualifiers) throws IOException {
		if (tableName == null || tableName.isEmpty() || row == null
				|| row.length <= 0 || qualifiers == null
				|| qualifiers.size() <= 0) {
			return;
		}

		HTableInterface table = null;

		try {
			table = HBaseTablePool.get(tableName);

			Put put = new Put(row);

			for (Entry<String, byte[]> qualifier : qualifiers.entrySet()) {
				put.add(Constants.HBASE_TABLE_FAMILY,
						Bytes.toBytes(qualifier.getKey()), qualifier.getValue());
			}

			table.put(put);
		} finally {
			if (table != null) {
				HBaseTablePool.put(table);
			}
		}
	}

	public static void putRows(String tableName, Row... rows)
			throws IOException {
		if (tableName == null || tableName.isEmpty() || rows == null) {
			return;
		}

		HTableInterface table = null;

		try {
			table = HBaseTablePool.get(tableName);

			List<Put> puts = new ArrayList<Put>();

			for (Row row : rows) {
				Put put = new Put(row.getRow());

				for (Entry<byte[], byte[]> entry : row
						.getQualifiersByByteArray().entrySet()) {
					put.add(Constants.HBASE_TABLE_FAMILY, entry.getKey(),
							entry.getValue());
				}

				puts.add(put);
			}

			table.put(puts);
		} finally {
			if (table != null) {
				HBaseTablePool.put(table);
			}
		}
	}

	public static void deleteRow(String tableName, byte[] row)
			throws IOException {
		if (tableName == null || tableName.isEmpty() || row == null
				|| row.length <= 0) {
			return;
		}

		HTableInterface table = null;

		try {
			table = HBaseTablePool.get(tableName);

			table.delete(new Delete(row));
		} finally {
			if (table != null) {
				HBaseTablePool.put(table);
			}
		}
	}

	public static void deleteRows(String tableName, List<byte[]> rows)
			throws IOException {
		if (tableName == null || tableName.isEmpty() || rows == null
				|| rows.size() <= 0) {
			return;
		}

		HTableInterface table = null;

		try {
			table = HBaseTablePool.get(tableName);

			List<Delete> deletes = new ArrayList<Delete>();

			for (byte[] row : rows) {
				deletes.add(new Delete(row));
			}

			table.delete(deletes);
		} finally {
			if (table != null) {
				HBaseTablePool.put(table);
			}
		}
	}

	public static void main(String[] args) throws Throwable {
		CrawlConfig config = new CrawlConfig();

		config.setHbaseMaster("leaderman.rain.centos:60010");
		config.setHbaseZooKeeperQuorum("leaderman.rain.centos");
		config.setHbaseZooKeeperPropertyClientPort("2181");
		config.setHbaseTablePoolSize(1);

		HBaseTablePool.initialize(config);

		System.out.println(HBaseUtil.getMaximumLongValueInFirstQualifier(
				Constants.HBASE_TABLE_DOCIDSERVER, null, null));

		HBaseTablePool.shutdown();
	}

}
